package com.atguigu.gmall.realtime.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.atguigu.gmall.realtime.bean.{PageActionLog, PageLog, StartLog}
import com.atguigu.gmall.realtime.util.{MyKafkaSink, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object BaseLohApp {
  // 1. 消费kafka主题
  // 2. 转换结构
  // 3. 分流
  //    启动主题，DWD_START_LOG  对应PageLog
  //    页面访问主题，DWD_PAGE_LOG
  //    页面动作主题，DWD_PAGE_ACTION
  //    页面曝光主题，DWD_PAGE_DISPLAY
  //    错误主题，DWD_ERROR_INFO

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("base_log_app").setMaster("local[4]") //4核 并行数  最好保证kafka的分区数一致
    // 1. 消费kafka主题
    //上下文  为批次处理 5秒钟
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val topic = "ODS_BASE_LOG"
    val groupId = "base_log_group"
    val inputDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    // inputDStream.map(_.value()).print(10000)
    // 2. 转换结构
    // 把kakfa的消息转换为结构化对象//1 通用的结构化对象 hashmap，jsonObject 2 专用的结构化对象 javabean / case class
    val JsonObjDStream: DStream[JSONObject] = inputDStream.map { record =>
      val jsonString: String = record.value()
      val JsonObj: JSONObject = JSON.parseObject(jsonString)
      JsonObj
    }

    // 3. 分流
    JsonObjDStream.foreachRDD {
      rdd =>
        rdd.foreach { jsonObj =>
          // 1） 提取页面访问主题的数据
          val commonObj = jsonObj.getJSONObject("common")
          val mid: String = commonObj.getString("mid")
          val uid: String = commonObj.getString("uid")
          val ar: String = commonObj.getString("ar")
          val ch: String = commonObj.getString("ch")
          val os: String = commonObj.getString("os")
          val md: String = commonObj.getString("md")
          val vc: String = commonObj.getString("vc")
          val isNew: String = commonObj.getString("is_new")
          val ts: Long = jsonObj.getLong("ts")
          val errorJsonObj: JSONObject = jsonObj.getJSONObject("err")
          val startJsonObj: JSONObject = jsonObj.getJSONObject("start")

          val pageObj: JSONObject = jsonObj.getJSONObject("page")
          if(errorJsonObj!=null){
            MyKafkaSink.send("dwd_error_info", jsonObj.toJSONString)
          }else if(startJsonObj!=null){
            //如果是错误日志 直接提取发送
            val entry =startJsonObj.getString("entry")
            val loading_time_ms =startJsonObj.getLong( "loading_time_ms")
            val open_ad_id =startJsonObj.getString("open_ad_id")
            val open_ad_ms =startJsonObj.getLong("open_ad_ms")
            val  open_ad_skip_ms =startJsonObj.getLong("open_ad_skip_ms")
            val startLog = StartLog(mid, uid, ar, ch, isNew, md, os, vc,  entry,open_ad_id,loading_time_ms ,open_ad_ms,open_ad_skip_ms,ts)
            MyKafkaSink.send("dwd_start_log",JSON.toJSONString(startLog,new SerializeConfig(true)))
          } else
          if (pageObj != null && pageObj.size() > 0) {
            val pageId = pageObj.getString("page_id")
            val pageItem = pageObj.getString("item")
            val pageItemType = pageObj.getString("item_type")
            val lastPageId = pageObj.getString("last_page_id")
            val duringTime = pageObj.getLong("during_time")
            val pageInfo = PageLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime, ts)
            // 发送到kafka指定的主题中
            MyKafkaSink.send("dwd_page_log", JSON.toJSONString(pageInfo, new SerializeConfig(true)))
            // 2） 提取页面动作主题的数据
            val actionJsonArr: JSONArray = jsonObj.getJSONArray("actions")
            if (actionJsonArr != null && actionJsonArr.size() > 0) {
              for (i<- 0 to actionJsonArr.size()-1) {
                val actionJsonObj: JSONObject = actionJsonArr.getJSONObject(i)
                val action_id = actionJsonObj.getString("action_id")
                val item = actionJsonObj.getString("item")
                val item_type = actionJsonObj.getString("item_type")
                val pageActionLog = PageActionLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime, action_id, item, item_type, ts)
                MyKafkaSink.send("dwd_page_action", JSON.toJSONString(pageActionLog, new SerializeConfig(true)))
              }
            }
          }
        }


    }
    ssc.start()
    ssc.awaitTermination()
  }

}